package cn.hexcloud.dga.governance.assessor.storage;

import cn.hexcloud.dga.governance.assessor.Assessor;
import cn.hexcloud.dga.governance.bean.AssessParam;
import cn.hexcloud.dga.governance.bean.GovernanceAssessDetail;
import cn.hexcloud.dga.governance.bean.GovernanceMetric;
import cn.hexcloud.dga.meta.bean.TableMetaInfo;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.hive.metastore.api.TableMeta;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Component("TABLE_SIMILAR")
public class TableSimilarAssessor extends Assessor {
    @Override
    public void checkProblem(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws Exception {
        TableMetaInfo curTableMetaInfo = assessParam.getTableMetaInfo();
        // 同层次才比较，ods层不参与比较
        String odsLevel = "ODS";
        if (odsLevel.equals(curTableMetaInfo.getTableMetaInfoExtra().getDwLevel())) {
            return;
        }
        GovernanceMetric governanceMetric = assessParam.getGovernanceMetric();
        // 取出相似度阈值
        BigDecimal percent = JSONObject.parseObject(governanceMetric.getMetricParamsJson()).getBigDecimal("percent").divide(BigDecimal.valueOf(100), 2, BigDecimal.ROUND_DOWN);
        List<TableMetaInfo> tableMetaInfoList = assessParam.getTableMetaInfoList();
        // 保存当前tableMetaInfo的所有字段名
        Set<String> curColNameSet = JSONObject.parseArray(curTableMetaInfo.getColNameJson(), JSONObject.class)
                .stream()
                .map(a -> a.getString("name")).collect(Collectors.toSet());
        // 储存相似度信息
        JSONArray similarityJSONArr = new JSONArray();
        for (TableMetaInfo otherTableMetaInfo : tableMetaInfoList) {
            // 跳过tableMetaInfoList中的当前tableMetaInfo
            if(otherTableMetaInfo.getSchemaName().equals(curTableMetaInfo.getSchemaName())&&otherTableMetaInfo.getTableName().equals(curTableMetaInfo.getTableName())){
                continue;
            }
            // 评判标准：当前表与其他表双方都要超过相似度阈值才算相似
            Set<String> otherColNameSet = JSONObject.parseArray(otherTableMetaInfo.getColNameJson(), JSONObject.class)
                    .stream()
                    .map(a -> a.getString("name")).collect(Collectors.toSet());

            BigDecimal curSimilarity = getSimilarity(curColNameSet, otherColNameSet);
            if (curSimilarity.compareTo(percent) < 0) {
                continue;
            }

            BigDecimal otherSimilarity = getOtherSimilarity(curColNameSet, otherColNameSet);
            if (otherSimilarity.compareTo(percent) >= 0) {
                JSONObject similarityJsonObj = new JSONObject();
                similarityJsonObj.put("otherTableName", otherTableMetaInfo.getSchemaName() + "." + otherTableMetaInfo.getTableName());
                similarityJsonObj.put("similarity", curSimilarity);
                similarityJsonObj.put("reverseSimilarity", otherSimilarity);
                similarityJSONArr.add(similarityJsonObj);
            }
        }
        if(similarityJSONArr.size()!=0){
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem(String.format("存在表相似度大于%.2f的表,详情如下:%s",percent,JSONArray.toJSONString(similarityJSONArr)));
        }
    }

    private BigDecimal getSimilarity(Set<String> curColNameSet, Set<String> otherColNameSet) {
        int similarCnt = 0;
        for (String curColName : curColNameSet) {
            if (otherColNameSet.contains(curColName)) {
                similarCnt++;
            }
        }
        return BigDecimal.valueOf(similarCnt).divide(BigDecimal.valueOf(curColNameSet.size()), 2, BigDecimal.ROUND_DOWN);
    }

    private BigDecimal getOtherSimilarity(Set<String> curColNameSet, Set<String> otherColNameSet) {
        return getSimilarity(otherColNameSet, curColNameSet);
    }
}
